﻿#include "logger\LogBroker.h"

#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <map>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

namespace Logging {
class LogBroker::Impl {
 public:
  // 构造函数：启动后台工作线程
  Impl() : done_(false) {
    worker_thread_ = std::thread(&Impl::WorkerLoop, this);
  }

  // 析构函数：确保线程被优雅地关闭和加入
  ~Impl() { Shutdown(); }

  void RegisterSink(const SinkPtr& sink, const LogCategory& category) {
    std::lock_guard<std::mutex> lock(sinks_mutex_);
    sinks_[category].push_back(sink);
  }

  void UnregisterSink(const SinkPtr& sink, const LogCategory& category) {
    std::lock_guard<std::mutex> lock(sinks_mutex_);
    auto it = sinks_.find(category);
    if (it != sinks_.end()) {
      auto& sink_vector = it->second;
      sink_vector.erase(
          std::remove(sink_vector.begin(), sink_vector.end(), sink),
          sink_vector.end());
    }
  }

  void Push(LogMessageInfo&& message) {
    auto task = [this, msg = std::move(message)]() {
      this->DeliverMessage(msg);
    };
    {
      std::lock_guard<std::mutex> lock(queue_mutex_);
      task_queue_.push(std::move(task));
    }  // 锁释放
    cv_.notify_one();  // 唤醒工作线程
  }

  void FlushAll() {
    auto task = [this]() { this->FlushSinks(); };
    {
      std::lock_guard<std::mutex> lock(queue_mutex_);
      task_queue_.push(std::move(task));
    }
    cv_.notify_one();
  }

  void Shutdown() {
    // 第一次调用时执行关闭流程
    if (!done_.exchange(true)) {
      cv_.notify_one();  // 确保线程能从等待中唤醒以检查 done_ 标志
      if (worker_thread_.joinable()) {
        worker_thread_.join();
      }
    }
  }

 private:
  // 后台线程的主循环
  void WorkerLoop() {
    while (!done_) {
      std::queue<std::function<void()>> local_queue;
      {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        cv_.wait(lock, [this] { return !task_queue_.empty() || done_; });
        if (done_ && task_queue_.empty()) return;
        local_queue.swap(task_queue_);
      }

      // 执行队列中的所有任务
      ProcessMessages(local_queue);
    }

    // 退出循环后，处理队列中可能剩余的最后一些消息
    DrainRemainingMessages();
  }

  void FlushSinks() {
    std::lock_guard<std::mutex> lock(sinks_mutex_);
    for (auto const& [category, sink_vec] : sinks_) {
      for (const auto& sink : sink_vec) {
        sink->Flush();
      }
    }
  }

  // 处理本地队列中的消息
  void ProcessMessages(std::queue<std::function<void()>>& queue) {
    while (!queue.empty()) {
      queue.front()();  // 执行任务
      queue.pop();
    }
  }

  // 清空主队列中所有剩余的消息
  void DrainRemainingMessages() {
    std::queue<std::function<void()>> local_queue;
    {
      std::unique_lock<std::mutex> lock(queue_mutex_);
      local_queue.swap(task_queue_);
    }
    ProcessMessages(local_queue);
  }

  // 将单条消息分发给所有匹配的 Sinks
  void DeliverMessage(const LogMessageInfo& message) {
    std::lock_guard<std::mutex> lock(sinks_mutex_);  // 访问sinks_需要加锁

    auto deliver_to = [&](const LogCategory& cat) {
      auto it = sinks_.find(cat);
      if (it != sinks_.end()) {
        for (const auto& sink : it->second) {
          if (sink) sink->Log(message);
        }
      }
    };

    deliver_to(message.category);
    if (message.category != "*") {  // 避免重复分发给通配符
      deliver_to("*");
    }
  }

  // 线程与同步原语
  std::atomic<bool> done_;
  std::thread worker_thread_;
  std::queue<std::function<void()>> task_queue_;
  std::mutex queue_mutex_;
  std::condition_variable cv_;

  // Sink 存储
  std::mutex sinks_mutex_;
  std::map<LogCategory, std::vector<SinkPtr>> sinks_;
};

// --- LogBroker 公共接口实现 ---
LogBroker::LogBroker() : pimpl_(std::make_unique<Impl>()) {}
LogBroker::~LogBroker() = default;  // PIMPL 需要在 cpp 中定义析构

LogBroker& LogBroker::GetInstance() {
  static LogBroker instance;
  return instance;
}
void LogBroker::RegisterSink(const SinkPtr& sink, const LogCategory& category) {
  pimpl_->RegisterSink(sink, category);
}
void LogBroker::UnregisterSink(const SinkPtr& sink,
                               const LogCategory& category) {
  pimpl_->UnregisterSink(sink, category);
}
void LogBroker::Push(LogMessageInfo&& message) {
  pimpl_->Push(std::move(message));
}
void LogBroker::FlushAll() { pimpl_->FlushAll(); }
void LogBroker::Shutdown() { pimpl_->Shutdown(); }
}  // namespace Logging